package com.kingsoft.dc.khaos.module.mpp.operator;

import com.kingsoft.dc.khaos.module.mpp.client.ConfVars;
import com.kingsoft.dc.khaos.module.mpp.client.KhaosContants;
import com.kingsoft.dc.khaos.module.mpp.client.MppClient;
import com.kingsoft.dc.khaos.module.mpp.exception.MppSqlException;
import com.kingsoft.dc.khaos.module.mpp.metadata.CombineStrategy;
import com.kingsoft.dc.khaos.module.mpp.metadata.MppConfig;
import com.kingsoft.dc.khaos.module.mpp.model.ColumnEntity;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.SqlBuilderUtils;
import com.kingsoft.dc.khaos.plugin.core.KhaosContext;

import java.util.List;

/**
 * 当前全量表.
 */
public class TyCurrentAll extends CombineStrategy {

    @Override
    public void execute(KhaosContext kc, String moduleId, MppConfig mppConfig) {
        //初始化MPP client,client已经初始化数据库连接
        MppClient client = null;
        try {
            client = new MppClient(mppConfig, kc);

            //获取操作的库表参数
            //比如获取源表所属的schema
            String srcSchema = String.format("\"%s\"", mppConfig.getSrcSchema());
            log().info("源库名srcSchema===>  " + srcSchema);
            //源表名
            String srcTable = String.format("\"%s\"", mppConfig.getSrcTable());
            log().info("源表名srcTable===>  " + srcTable);
            //源表过滤条件
            String filter = mppConfig.getDataFilter();

            //目标表所属schema名
            String tarSchema = String.format("\"%s\"", mppConfig.getTarSchema());
            log().info("目标库名tarSchema===>  " + tarSchema);
            //目标表名
            String tarTable = String.format("\"%s\"", mppConfig.getTarTable());
            log().info("目标表名tarTable===>  " + tarTable);
            //主键列表
            List<String> keys = mppConfig.getpKeys();

            //获取源档类型
            String src_Type = mppConfig.getSrcType();

            //获取主键字符串
            String pKeys = SqlBuilderUtils.selKeys(tarTable, keys);
            log().info("目标表主键字符串===>  " + pKeys);

            //是否物理删除
            boolean isDeleted = mppConfig.isDeleted();

            //获取追加字段名
            //比如获取[数据日期]字段名
            String bizDateColName = client.getAppendColName(ConfVars.P9_DATA_DATE);

            //获取追加字段值，比如获取[数据日期]字段值
            String bizDate = kc.getConf().getString(KhaosContants.BIZ_DATE);
            String batchNo = kc.getConf().getString(KhaosContants.BATCH_NO);
            String jobName = kc.getConf().getString(KhaosContants.JOB_NAME);
            String jobId = kc.getConf().getString(KhaosContants.JOB_INSTANCE_ID);

            //获取表全字段信息
//            List<ColumnEntity> columnEntities = client.getColumnInfo(srcSchema, srcTable);
            String cols = client.getColumnNames(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            log().info("源表全字段信息===>  " + cols);
            //拼接当前全量表算子SQL
            String exe_sql = buildSql(src_Type, cols, pKeys, tarSchema, tarTable, srcSchema, srcTable, bizDate, jobName, batchNo, client, filter);
            log().info("当前全量表算子SQL===>  " + exe_sql);

            client.execInsert(exe_sql);
        } catch (Exception e) {
            e.printStackTrace();
            throw new MppSqlException("执行MPP SQL 异常");
        } finally {
            if (client != null) {
                client.close();
            }
        }

    }

    //拼接当前全量表算子SQL实现类
    public String buildSql(String src_Type,
                           String cols,
                           String pKeys,
                           String tarSchema,
                           String tarTable,
                           String srcSchema,
                           String srcTable,
                           String bizDate,
                           String jobName,
                           String batchNo,
                           MppClient client,
                           String filter) {
        log().info("TyCurrentAll buildSql start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
        String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String tyRecInitDate = client.getAppendColName(ConfVars.P9_REC_INIT_DATE);
        String dataFilter = "";
        if (null != filter && !filter.equals("")) {
            dataFilter = "WHERE " + filter;
        }
        String exe_sql = "";
        if (src_Type.equalsIgnoreCase("ALL")) {
            exe_sql = String.format("BEGIN ; \n DELETE FROM %s.%s ;", tarSchema, tarTable);
            String union = String.format("INSERT INTO  %s.%s (%s", tarSchema, tarTable, cols)
                    + "," + tyDataDate + ", " + tyJobName + "," + tyBatchNumber + "," + tyDelFlag + "," + tyDelDate + "," + tyDelBatch + "," + tyRecInitDate + " ) \n "
                    + String.format("select %s,to_date('%s','YYYYMMDD') as " + tyDataDate + ",'%s' as " + tyJobName + ",'%s' as " + tyBatchNumber + ",'0' AS " + tyDelFlag + " \n ", cols, bizDate, jobName, batchNo)
                    + String.format(",to_date('00010101','YYYYMMDD') as " + tyDelDate + ",'0000' AS " + tyDelBatch + ",to_date('%s','YYYYMMDD') as ty_rec_init_date \n ", bizDate)
                    + String.format("from %s.%s %s; \n COMMIT ; \n ", srcSchema, srcTable, dataFilter);
            exe_sql = exe_sql + "\n" + union;
        }
        log().info("TyCurrentAll buildSql end...");
        return exe_sql;
    }
}
